#include "Bitcask.h"
#include <cstring>
#include <cstdlib>
#include <cstdio>
#include <cerrno>
#include <unistd.h>
#include <fcntl.h>
#include <dirent.h>
#include <algorithm>
#include <sys/stat.h>
#include <pthread.h>
#include "BTreeIndex.h"
#include "Objects.h"
#include "Log.h"

using namespace std;

int Bitcask::Init(const char *path, BtreeIndex *index)
{
    char tmp_path[PATH_MAX];
    bool need_recovery = true;
    bool new_db = false;

    my_index = index;
    key_buf_pos = val_buf_pos = 0;
    buf_key_num = 0;

    datafiles.clear();

    if (strlen(path) > PATH_MAX - 2)
    {
        LOG_ERROR("DB directory name [%s] too long", path);
        return -1;
    }

    if (path[strlen(path)-1] == '/')
        snprintf(data_path, PATH_MAX, "%s", path);
    else
        snprintf(data_path, PATH_MAX, "%s/", path);

    snprintf(tmp_path, PATH_MAX, "%smanifest", data_path);

    for(;;)
    {
        DataFile *df_ptr;
        int fd;
        u32 valid_df_num = 0;

        fd = open(tmp_path, O_RDONLY);
        if (fd < 0)
        {
            LOG_WARN("DB manifest [%s] not found.", tmp_path);
            break;
        }


        if (read(fd, &valid_df_num, sizeof(u32)) == 0)
        {
            LOG_WARN("DB manifest read error");
            break;
        }

        for(u32 i = 0; i < valid_df_num; ++i)
        {
            df_ptr = (DataFile*)mymalloc(sizeof(DataFile));
            if (read(fd, df_ptr, sizeof(DataFile)) != sizeof(DataFile))
            {
                LOG_WARN("DB manifest read error");
                break;
            }

            if (df_ptr->idx != datafiles.size())
            {
                while(datafiles.size() < df_ptr->idx)
                    datafiles.push_back(NULL);
            }
            datafiles.push_back(df_ptr);
        }
        

        if (load_index() < 0)
        {
            LOG_WARN("DB reload index error");
            for(int i = 0; i < datafiles.size(); ++i)
                if (datafiles[i] != NULL) myfree(datafiles[i]);
            datafiles.clear();
            break;
        } 

        need_recovery = false;
        break;
    }

    unlink(tmp_path);

    if (need_recovery)
    {
        LOG_WARN("DB manifest invalid, try to replay db files");
        if (recovery_from_df() < 0)
        {
            LOG_ERROR("Recovery faild.");
            return -1;
        }
    }

    cur_seq = 0;
    cur_idx = 0;
    cur_seq_history = 0;
    for(off_t i = 0; i < datafiles.size(); ++i)
    {
        if (datafiles[i] != NULL && datafiles[i]->seq > cur_seq)
        {
            cur_seq = datafiles[i]->seq;
            cur_idx = i;
        }
        if (datafiles[i] != NULL && datafiles[i]->seq_history > cur_seq_history)
            cur_seq_history = datafiles[i]->seq_history;
    }

    if (datafiles.size() == 0)
    {
        DataFile *df_ptr = (DataFile*)mymalloc(sizeof(DataFile));
        memset(df_ptr, 0, sizeof(DataFile));
        datafiles.push_back(df_ptr);
        new_db = true;
    }

    snprintf(tmp_path, PATH_MAX, "%s%09d.key", data_path, cur_seq);
    cur_fd_key = open(tmp_path, O_WRONLY | O_APPEND | O_CREAT, S_IRUSR | S_IWUSR | S_IRGRP | S_IROTH);
    if (cur_fd_key < 0)
    {
        LOG_ERROR("Open data file %s for write failed", tmp_path);
        return -1;
    }
    snprintf(tmp_path, PATH_MAX, "%s%09d.val", data_path, cur_seq);
    cur_fd_val = open(tmp_path, O_WRONLY | O_APPEND | O_CREAT, S_IRUSR | S_IWUSR | S_IRGRP | S_IROTH);
    if (cur_fd_val < 0)
    {
        LOG_ERROR("Open data file %s for write failed", tmp_path);
        return -1;
    }

    if (new_db)
    {
        char header_buf[8];
        *(u32*)(&header_buf[0]) = 0;
        *(u32*)(&header_buf[4]) = 0;
        write(cur_fd_key, header_buf, 8);
    }
    
    key_buf = (char*)mymalloc(BUF_SIZE);
    val_buf = (char*)mymalloc(BUF_SIZE);
    key_buf_swap = (char*)mymalloc(BUF_SIZE);
    val_buf_swap = (char*)mymalloc(BUF_SIZE);
    lock.Init();
    compact_status = 0;
    is_running = true;

    if (pthread_create(&aux_tid, NULL, Bitcask::aux_thread, this))
    {
        LOG_ERROR("DB create auxillary thread failed: %s", strerror(errno));
        return -1;
    }
    return 0;
}

void Bitcask::Close()
{
    int cur_fd;
    char tmp_path[PATH_MAX];
    char write_buf[8192];
    u32 write_buf_pos;
    void *tret;

    BtreeBucket *trav_tree;
    u32 valid_df_num = 0;

    is_running = false;
    pthread_join(aux_tid, &tret);

    WriteOut();
    fdisksync();
    if (cur_fd_key >= 0) close(cur_fd_key);
    if (cur_fd_val >= 0) close(cur_fd_val);

    snprintf(tmp_path, PATH_MAX, "%stree.hint", data_path);
    cur_fd = open(tmp_path, O_WRONLY | O_TRUNC | O_CREAT, S_IRUSR | S_IWUSR | S_IRGRP | S_IROTH);
    if (cur_fd < 0)
    {
        LOG_ERROR("Open index hint file %s for write failed", tmp_path);
        return;
    }

    trav_tree = my_index->leaf_head;
    write_buf_pos = 4;
    while(trav_tree != NULL)
    {
        for(int i = 0; i < trav_tree->element_num; ++i)
        {
            Item *cur_item = reinterpret_cast<Item*>(trav_tree->ptrs[i]);
            if (write_buf_pos + 12 + cur_item->KeyLength() > 8192)
            {
                *(u32*)(&write_buf[0]) = write_buf_pos - 4;
                write(cur_fd, write_buf, write_buf_pos);
                write_buf_pos = 4;
            }
            *(u32*)(&write_buf[write_buf_pos]) = cur_item->len;
            write_buf_pos += 4;
            *(u64*)(&write_buf[write_buf_pos]) = cur_item->real_pos;
            write_buf_pos += 8;
            memcpy(&write_buf[write_buf_pos], &cur_item->data[0], cur_item->KeyLength());
            write_buf_pos += cur_item->KeyLength();
        }
        trav_tree = trav_tree->next_leaf;
    }

    if (write_buf_pos)
    {
        *(u32*)(&write_buf[0]) = write_buf_pos - 4;
        write(cur_fd, write_buf, write_buf_pos);
        *(u32*)(&write_buf[0]) = 0x19890604;
        write(cur_fd, write_buf, 4);
        write_buf_pos = 0;
    }
    close(cur_fd);

    snprintf(tmp_path, PATH_MAX, "%smanifest", data_path);
    cur_fd = open(tmp_path, O_WRONLY | O_TRUNC | O_CREAT, S_IRUSR | S_IWUSR | S_IRGRP | S_IROTH);
    if (cur_fd < 0)
    {
        LOG_ERROR("Open manifest file %s for write failed", tmp_path);
        return;
    }
    for(int i = 0; i < datafiles.size(); ++i)
        if (datafiles[i] != NULL) valid_df_num++;

    write(cur_fd, &valid_df_num, 4);
    
    for(int i = 0; i < datafiles.size(); ++i)
        if (datafiles[i] != NULL) write(cur_fd, datafiles[i], sizeof(DataFile));

    close(cur_fd);

    return;
}


void Bitcask::Destroy()
{
    lock.Destroy();
    for(u32 i = 0; i < df_num; ++i)
        myfree(datafiles[i]);
    datafiles.clear();
    myfree(key_buf);
    myfree(val_buf);
    myfree(val_buf_swap);
    myfree(key_buf_swap);
}

int Bitcask::load_index()
{
    int cur_fd;
    char tmp_path[PATH_MAX];
    char read_buf[8192];

    snprintf(tmp_path, PATH_MAX, "%stree.hint", data_path);
    cur_fd = open(tmp_path, O_RDONLY);
    lseek(cur_fd, 0, SEEK_SET);

    if (cur_fd < 0)
    {
        LOG_ERROR("Open index file %s for read failed", tmp_path);
        return -1;
    }

    for(;;)
    {
        u32 buf_size;
        int x = read(cur_fd, &buf_size, 4);
        if (x != 4)
        {
            //MyIndex.clear;
            LOG_ERROR("Read index file %s error", tmp_path);
            close(cur_fd);
            return -1;
        }
        if (buf_size == 0x19890604) break;

        if (buf_size > 8192 || read(cur_fd, read_buf, buf_size) != buf_size)
        {
            //MyIndex.clear;
            LOG_ERROR("Read index file %s error", tmp_path);
            close(cur_fd);
            return -1;
        }

        for(u32 pos = 0; pos < buf_size;)
        {
            u32 key_len = *(u32*)(&read_buf[pos]) >> 24;
            Item *cur_item = (Item*)mymalloc(sizeof(Item) + key_len);
            cur_item->Init(*(u32*)(&read_buf[pos]));
            cur_item->real_pos = *(u64*)(&read_buf[pos+4]);
            memcpy(&cur_item->data[0], &read_buf[pos+12], key_len);
            pos += 12 + key_len;
            my_index->Set(cur_item);
        }
    }
    close(cur_fd);
    unlink(tmp_path);

    return 0;
}


int Bitcask::recovery_from_df()
{
    struct dirent **namelist;
    int n;

    n = scandir(data_path, &namelist, 0, alphasort);
    if (n < 0)
    {
        LOG_ERROR("scan data file in directory %s error", data_path);
        return -1;
    }

    for(int i = 0; i < n; ++i) 
    {
        int seq = atoi(namelist[i]->d_name);
        char tmp_filename[16];

        snprintf(tmp_filename, 16, "%09d.key", seq);
        for(;;)
        {
            char tmp_path[PATH_MAX];
            char read_buf[8192];
            int buf_pos, buf_size;
            int fd;
            u32 cur_idx, cur_seq_history;
            DataFile *df_ptr;
            
            if (strlen(namelist[i]->d_name) != 13 
                || memcmp(tmp_filename, namelist[i]->d_name, 13) != 0)
                break;

            LOG_INFO("Found data file [%s%s]", data_path, tmp_filename);
            snprintf(tmp_path, PATH_MAX, "%s%s", data_path, tmp_filename);
            fd = open(tmp_path, O_RDONLY);
            if (fd < 0)
            {
                LOG_WARN("DB key file [%s] open failed", tmp_path);
                break;
            }

            if (read(fd, read_buf, 8) != 8)
            {
                close(fd);
                break;
            }
            
            cur_idx = *(u32*)(&read_buf[0]);
            cur_seq_history = *(u32*)(&read_buf[4]);

            df_ptr = (DataFile*)mymalloc(sizeof(DataFile));
            df_ptr->idx = cur_idx;
            df_ptr->seq = seq;
            df_ptr->seq_history = cur_seq_history;
            df_ptr->key_size = 8;
            df_ptr->count = 0;
            df_ptr->val_size = 0;

            if (datafiles.size() <= df_ptr->idx)
            {
                while(datafiles.size() < df_ptr->idx)
                    datafiles.push_back(NULL);
                datafiles.push_back(df_ptr);
            }
            else
            {
                datafiles[df_ptr->idx] = df_ptr;
            }

            buf_pos = buf_size = 0;
            Item *del_item = (Item*)mymalloc(sizeof(Item) + 256);
            for(bool eof = false;;)
            {
                u8 cur_key_len;
                Item *item;

                if (buf_size - buf_pos < 13
                    || buf_size - buf_pos < (*(u32*)(&read_buf[buf_pos+1]) >> 24) + 13)
                {
                    int read_size;
                    if (eof) break;
                    memmove(&read_buf[0], &read_buf[buf_pos], buf_size - buf_pos);
                    buf_size = buf_size - buf_pos;
                    buf_pos = 0;
                    if ((read_size = read(fd, &read_buf[buf_size], 8192 - buf_size)) != 8192 - buf_size)
                        eof = true;
                    buf_size += read_size;
                }

                if (buf_size - buf_pos < 13
                    || buf_size - buf_pos < (*(u32*)(&read_buf[buf_pos+1]) >> 24) + 13)
                {
                    break;
                }

                cur_key_len = *(u32*)(&read_buf[buf_pos+1]) >> 24;
                switch(read_buf[buf_pos])
                {
                case INSERT:
                case UPDATE:
                    item = (Item*)mymalloc(sizeof(Item)+cur_key_len);
                    memcpy(&(item->data[0]), &read_buf[buf_pos+13], cur_key_len);
                    item->Init(*(u32*)(&read_buf[buf_pos+1]));
                    item->real_pos = ((u64)cur_idx << 32)| *(u32*)(&read_buf[buf_pos+5]);
                    if (my_index->Set(item,true) == NULL)
                    {
                        u32 update_idx = item->real_pos >> 32;
                        if (datafiles[update_idx] != NULL)
                        {
                            datafiles[update_idx]->garbage_count++;
                            datafiles[update_idx]->garbage_size += *(u32*)(&read_buf[buf_pos+1]) & 0x00ffffff;
                            datafiles[update_idx]->garbage_size += cur_key_len + 13;
                        }
                        else
                        {
                            LOG_WARN("Internal error. data_status unmatch");
                        }
                        myfree(item);
                    }
                    df_ptr->key_size += 13 + cur_key_len;
                    df_ptr->val_size += *(u32*)(&read_buf[buf_pos+1]) & 0x00ffffff;;
                    df_ptr->count++;
                    break;
                case DELETE:
                    del_item->Init(*(u32*)(&read_buf[buf_pos+1]));
                    memcpy(&(del_item->data[0]), &read_buf[buf_pos+13], cur_key_len);
                    item = my_index->Delete(del_item);
                    if (item == NULL)
                    {
                        u32 del_idx = *(u32*)&read_buf[buf_pos+5];
                        if (del_idx < datafiles.size() && datafiles[del_idx] != NULL 
                            && datafiles[del_idx]->seq_history == *(u32*)&read_buf[buf_pos+9])
                        {
                            datafiles[del_idx]->garbage_count++;
                            datafiles[del_idx]->garbage_size += *(u32*)(&read_buf[buf_pos+1]) & 0x00ffffff;
                        }
                    }
                    else
                    {
                        item->Delete();
                    }
                    df_ptr->count++;
                    break;
                default:
                    LOG_ERROR("Data file [%s] invalid.", tmp_path);
                    eof = true;
                    break;
                }
                buf_pos += cur_key_len + 13;
            }
            myfree(del_item);
            close(fd);
            break;
        }
        free(namelist[i]);
    }
    free(namelist);
    return 0;
}


u64 Bitcask::Update(char *key, int key_size, char *val, int val_size, 
                    u64 old_pos, u32 old_size, bool fsync)
{
    u32 old_idx = old_pos >> 32;
    u32 old_off = old_pos & 0xffffffff;
    char item_header[13];
    
    item_header[0] = UPDATE;
    *(u32*)(&item_header[1]) = ((u32)key_size << 24)| (u32)val_size;

    lock.WLock();
    datafiles[old_idx]->garbage_count++;
    datafiles[old_idx]->garbage_size += (old_size & 0x00ffffff) + (old_size >> 24);

    if (fsync || key_buf_pos + key_size + 5> BUF_SIZE || val_buf_pos + val_size > BUF_SIZE)
    {
        write_out();
    }

    *(u32*)(&item_header[5]) = datafiles[cur_idx]->val_size + val_buf_pos;

    if (fsync || val_size > BUF_SIZE)
    {
        write(cur_fd_key, &item_header[0], 13);
        write(cur_fd_key, key, key_size);
        write(cur_fd_val, val, val_size);
        datafiles[cur_idx]->key_size += 13 + key_size;
        datafiles[cur_idx]->val_size += val_size;
        datafiles[cur_idx]->count++;
    }
    else
    {
        memcpy(key_buf + key_buf_pos, &item_header[0], 13);
        key_buf_pos += 13;
        memcpy(key_buf + key_buf_pos, key, key_size);
        key_buf_pos += key_size;
        memcpy(val_buf + val_buf_pos, val, val_size);
        val_buf_pos += val_size;
        buf_key_num++;
    }
    lock.UnLock();
}


u64 Bitcask::Put(char *key, int key_size, char *val, int val_size, bool fsync)
{
    char item_header[13];
    u64 res = 0;

    item_header[0] = INSERT;
    *(u32*)(&item_header[1]) = ((u32)key_size << 24)| (u32)val_size;

    lock.WLock();

    if (fsync || key_buf_pos + key_size + 13> BUF_SIZE || val_buf_pos + val_size > BUF_SIZE)
    {
        write_out();
    }

    *(u32*)(&item_header[5]) = datafiles[cur_idx]->val_size + val_buf_pos;
	*(u32*)(&item_header[9]) = 0; //valid only in DELETE 

    if (fsync || val_size > BUF_SIZE)
    {
        write(cur_fd_key, &item_header[0], 13);
        write(cur_fd_key, key, key_size);
        write(cur_fd_val, val, val_size);
        res = ((u64)cur_idx << 32)| *(u32*)(&item_header[5]);
        datafiles[cur_idx]->key_size += 13 + key_size;
        datafiles[cur_idx]->val_size += val_size;
        datafiles[cur_idx]->count++;
    }
    else
    {
        memcpy(key_buf + key_buf_pos, &item_header[0], 13);
        key_buf_pos += 13;
        memcpy(key_buf + key_buf_pos, key, key_size);
        key_buf_pos += key_size;
        res = ((u64)cur_idx << 32)| *(u32*)(&item_header[5]);
        memcpy(val_buf + val_buf_pos, val, val_size);
        val_buf_pos += val_size;
        buf_key_num++;
    }
    lock.UnLock();  
    return res;
}

int Bitcask::Get(char *val, int val_size, u64 pos)
{
    u32 val_idx = pos >> 32;
    u32 val_off = pos & 0xffffffff;
    int cur_fd;
    char file_path[PATH_MAX];

    lock.RLock();
    if (val_idx == cur_idx && val_off >= datafiles[cur_idx]->val_size)
    {
        memcpy(val, val_buf + (val_off - datafiles[cur_idx]->val_size), val_size);
        lock.UnLock();
        return 0;
    }
    lock.UnLock();
    
    snprintf(file_path, PATH_MAX, "%s%09d.val", 
        data_path, (datafiles[val_idx]!=NULL?datafiles[val_idx]->seq:999999999));
    cur_fd = open(file_path, O_RDONLY);
    if (cur_fd < 0)
    {
        LOG_ERROR("Open data file for read error! %s%09d.val", 
            data_path, (datafiles[val_idx]!=NULL?datafiles[val_idx]->seq:999999999));
        return -1;
    }

    if (pread(cur_fd, val, val_size, val_off) != val_size)
    {
        LOG_ERROR("Read data file error! %s%09d.key/val", 
            data_path, (datafiles[val_idx]!=NULL?datafiles[val_idx]->seq:999999999));
        close(cur_fd);
        return -1;
    }

    close(cur_fd);
    return 0;
}


void Bitcask::Delete(char *key, int key_size, int val_size, u64 pos, bool fsync)
{
    u32 old_idx = pos >> 32;
    u32 old_off = pos & 0xffffffff;
    char item_header[13];

    item_header[0] = DELETE;
    *(u32*)(&item_header[1]) = ((u32)key_size << 24)| (u32)val_size;

    lock.WLock();
    if (datafiles[old_idx] != NULL)
    {
        datafiles[old_idx]->garbage_count++;
        datafiles[old_idx]->garbage_size += val_size;
    }
    
    if (fsync || key_buf_pos + key_size + 13> BUF_SIZE)
    {
        write_out();
    }

	*(u32*)(&item_header[5]) = datafiles[old_idx]->idx;
	*(u32*)(&item_header[9]) = datafiles[old_idx]->seq_history;

    if (fsync)
    {
        write(cur_fd_key, &item_header[0], 13);
        write(cur_fd_key, key, key_size);
        datafiles[cur_idx]->key_size += 13 + key_size;
        datafiles[cur_idx]->count++;
    }
    else
    {
        memcpy(key_buf + key_buf_pos, &item_header[0], 13);
        key_buf_pos += 13;
        memcpy(key_buf + key_buf_pos, key, key_size);
        key_buf_pos += key_size;
        buf_key_num++;
    }
    lock.UnLock();
}

void Bitcask::write_out()
{
    if (key_buf_pos == val_buf_pos && key_buf_pos == 0) return;
    write(cur_fd_key, key_buf, key_buf_pos);
    write(cur_fd_val, val_buf, val_buf_pos);
    datafiles[cur_idx]->key_size += key_buf_pos;
    datafiles[cur_idx]->val_size += val_buf_pos;
    datafiles[cur_idx]->count += buf_key_num;
    buf_key_num = key_buf_pos = val_buf_pos = 0;

    if (datafiles[cur_idx]->count > FILERECORD_LIMIT 
        || datafiles[cur_idx]->key_size + datafiles[cur_idx]->val_size > FILESIZE_LIMIT)
    {
        LOG_INFO("cur_idx: %u   cur_count:  %u  cur_key_size:%u cur_val_size: %u",
            cur_idx, datafiles[cur_idx]->count, datafiles[cur_idx]->key_size, datafiles[cur_idx]->val_size);
        bool reuse_id = false;
        char tmp_path[PATH_MAX];
        char header_buf[8];
        DataFile *df_ptr;
     
        if (cur_fd_key >= 0) close(cur_fd_key);
        if (cur_fd_val >= 0) close(cur_fd_val);

        for(size_t i = 0; i < datafiles.size(); ++i)
        {
            if (datafiles[i] == NULL)
            {
                cur_idx = i;
                reuse_id = true;
                break;
            }
        }
        
        if (!reuse_id)
        {
            cur_idx = datafiles.size();
            datafiles.push_back(NULL);
        }

        df_ptr = (DataFile*)mymalloc(sizeof(DataFile));
        memset(df_ptr, 0, sizeof(DataFile));
        
        cur_seq++;
        cur_seq_history++;

        df_ptr->idx = cur_idx;
        df_ptr->seq = cur_seq;
        df_ptr->seq_history = cur_seq_history;
        datafiles[cur_idx] = df_ptr;   

        LOG_INFO("New data file: idx: %09d seq: %09d", cur_idx, cur_seq);
    
        snprintf(tmp_path, PATH_MAX, "%s%09d.key", data_path, cur_seq);
        cur_fd_key = open(tmp_path, O_WRONLY | O_APPEND | O_CREAT, S_IRUSR | S_IWUSR | S_IRGRP | S_IROTH);
        snprintf(tmp_path, PATH_MAX, "%s%09d.val", data_path, cur_seq);
        cur_fd_val = open(tmp_path, O_WRONLY | O_APPEND | O_CREAT, S_IRUSR | S_IWUSR | S_IRGRP | S_IROTH);

        if (cur_fd_key < 0 || cur_fd_val < 0)
        {
            LOG_ERROR("Create new data file error! %s%09d.key/val", data_path, cur_seq);
            return;
        }
        *(u32*)(&header_buf[0]) = cur_idx;
        *(u32*)(&header_buf[4]) = cur_seq_history;
        write(cur_fd_key, header_buf, 8);
        datafiles[cur_idx]->key_size = 8;
    }
}


class IdxSeqComparator
{
public:
    IdxSeqComparator(Bitcask* bitcask):cur_bitcask(bitcask) {}
    bool operator ()(const u32 &a, const u32 &b)
    {
        return cur_bitcask->datafiles[a]->seq < cur_bitcask->datafiles[b]->seq;
    }
private:
    Bitcask* cur_bitcask;
};

void* Bitcask::compact(void* instance)
{
    Bitcask *bitcask = reinterpret_cast<Bitcask*>(instance);

    vector<u32> idx_list;
    
    bitcask->compact_status = 2;

    bitcask->lock.RLock();
    for(int i = 0; i < bitcask->datafiles.size(); ++i)
        if (bitcask->datafiles[i] != NULL && i != bitcask->cur_idx) idx_list.push_back(i);

    sort(idx_list.begin(), idx_list.end(), IdxSeqComparator(bitcask));
    bitcask->lock.UnLock();

    u32 cum_file_size = 0;
    u32 cum_item_count = 0;
    u32 cum_start_idx = 0;

    for(int i = 0; i < idx_list.size(); ++i)
    {
        bitcask->lock.RLock();
        if (bitcask->datafiles[idx_list[i]]->garbage_count > (FILERECORD_LIMIT >> 1)
            || bitcask->datafiles[idx_list[i]]->garbage_size > (FILESIZE_LIMIT >> 1))
        {
            if (cum_file_size == 0) cum_start_idx = i;
            cum_file_size += bitcask->datafiles[idx_list[i]]->val_size + bitcask->datafiles[idx_list[i]]->key_size;
        }
        else if (cum_file_size > 0)
        {
            cum_file_size += bitcask->datafiles[idx_list[i]]->val_size + bitcask->datafiles[idx_list[i]]->key_size;
            cum_item_count += bitcask->datafiles[idx_list[i]]->count - bitcask->datafiles[idx_list[i]]->garbage_count;
        }
        bitcask->lock.UnLock();

        if (cum_file_size >= FILESIZE_LIMIT || cum_item_count >= FILERECORD_LIMIT)
        {
            bitcask->compact_internal(&idx_list[cum_start_idx], i - cum_start_idx + 1);
            cum_file_size = 0;
            cum_item_count = 0;
        }
    }

    bitcask->compact_status = 1;
    return NULL;
}


void Bitcask::compact_internal(u32 *idx_list, int count)
{
    char tmp_path[PATH_MAX], tmp_path2[PATH_MAX];
    char header_buf[8];
    u32 new_seq, new_seq_history, new_idx;
    int new_fd_key, new_fd_val, old_fd_key, old_fd_val;
    u32 cur_key_pos, cur_val_pos;
    DataFile* new_df_ptr;

    lock.WLock();
    cur_seq_history++;
    new_seq = datafiles[idx_list[0]]->seq;
    new_seq_history = cur_seq_history;
    lock.UnLock();
    new_idx = idx_list[0];

    snprintf(tmp_path, PATH_MAX, "%s%09d.key.tmp", data_path, new_seq);
    new_fd_key = open(tmp_path, O_WRONLY | O_TRUNC | O_CREAT , S_IRUSR | S_IWUSR | S_IRGRP | S_IROTH);
    snprintf(tmp_path, PATH_MAX, "%s%09d.val.tmp", data_path, new_seq);
    new_fd_val = open(tmp_path, O_WRONLY | O_TRUNC | O_CREAT , S_IRUSR | S_IWUSR | S_IRGRP | S_IROTH);
    if (new_fd_key < 0 || new_fd_val < 0)
    {
        if (new_fd_key >= 0) close(new_fd_key);
        if (new_fd_val >= 0) close(new_fd_val);
        LOG_ERROR("Create temporary data file error! %s%09d.key.tmp/val.tmp", data_path, new_seq);
        return;
    }

    *(u32*)(&header_buf[0]) = new_idx;
    *(u32*)(&header_buf[4]) = new_seq_history;
    write(new_fd_key, header_buf, 8);

    new_df_ptr = (DataFile*)mymalloc(sizeof(DataFile));
    new_df_ptr->idx = new_idx;
    new_df_ptr->seq = datafiles[idx_list[0]]->seq;
    new_df_ptr->seq_history = new_seq_history;
    new_df_ptr->key_size = 8;
    new_df_ptr->count = 0;
    new_df_ptr->val_size = 0;
    new_df_ptr->garbage_count = 0;
    new_df_ptr->garbage_size = 0;
    cur_key_pos = 8;
    cur_val_pos = 0;


    for(int i = 0; i < count; ++i)
    {
        int old_idx = idx_list[i];

        lock.RLock();
        if (datafiles[old_idx] == NULL)
        {
            LOG_WARN("Try to compact idx %u, but didn't found info of this file.", idx_list[i]);
            lock.UnLock();
            return;
        }
        snprintf(tmp_path, PATH_MAX, "%s%09d.key", data_path, datafiles[old_idx]->seq);
        old_fd_key = open(tmp_path, O_RDONLY);
        snprintf(tmp_path, PATH_MAX, "%s%09d.val", data_path, datafiles[old_idx]->seq);
        old_fd_val = open(tmp_path, O_RDONLY);
        lock.UnLock();

        if (old_fd_key < 0 || old_fd_val < 0)
        {
            if (old_fd_key >= 0) close(old_fd_key);
            if (old_fd_val >= 0) close(old_fd_val);
            LOG_ERROR("Open data file error! %s%09d.key/val", data_path, new_seq);
            break;
        }

        u32 buf_pos = 0, buf_size = 0;
        char read_buf[8192];
        char move_buf[8192];
        Item *tmp_item = (Item*)mymalloc(sizeof(Item) + 256);

        if (read(old_fd_key, &read_buf[0], 8) == 8)
        for(bool eof = false;;)
        {
            u8 cur_key_len;
            u32 del_idx;
            Item *dst;

            if (buf_size < 13 + buf_pos
                || buf_size < (*(u32*)(&read_buf[buf_pos+1]) >> 24) + 13 + buf_pos)
            {
                int read_size;
                if (eof) break;
                memmove(&read_buf[0], &read_buf[buf_pos], buf_size - buf_pos);
                buf_size = buf_size - buf_pos;
                buf_pos = 0;
                if ((read_size = read(old_fd_key, &read_buf[buf_size], 8192 - buf_size)) != 8192 - buf_size)
                    eof = true;
                buf_size += read_size;
            }

            if (buf_size < 13 + buf_pos
                || buf_size < (*(u32*)(&read_buf[buf_pos+1]) >> 24) + 13 + buf_pos)
                break;

            cur_key_len = *(u32*)(&read_buf[buf_pos+1]) >> 24;
            switch(read_buf[buf_pos])
            {
            case INSERT:
            case UPDATE:
                memcpy(&(tmp_item->data[0]), &read_buf[buf_pos+13], cur_key_len);
                tmp_item->Init(*(u32*)(&read_buf[buf_pos+1]));
                tmp_item->real_pos = ((u64)old_idx << 32)| *(u32*)(&read_buf[buf_pos+5]);

                dst = my_index->Get(tmp_item);
                if (dst != NULL && dst->real_pos == tmp_item->real_pos)
                {
                    u32 val_size = dst->ValLength();
                    u32 old_pos = dst->real_pos & 0xffffffff;
                    //Move val into new;
                    dst->real_pos = ((u64)new_idx << 32) | cur_val_pos;
                    *(u32*)(&read_buf[buf_pos+5]) = cur_val_pos;
                    while(val_size > 0)
                    {
                        int tsize = MIN(val_size, 8192);
                        pread(old_fd_val, move_buf, tsize, old_pos);
                        write(new_fd_val, move_buf, tsize);
                        cur_val_pos += tsize;
                        val_size -= tsize;
                    }
                    write(new_fd_key, &read_buf[buf_pos], cur_key_len + 13);
                    cur_key_pos += cur_key_len + 13;
                    dst->ReleaseRef();
                    new_df_ptr->count++;
                }
                else
                {
                    //skip and ignore
                }
                break;
            case DELETE:
                del_idx = *(u32*)(&read_buf[buf_pos + 5]);
                if (del_idx < datafiles.size() && datafiles[del_idx] != NULL
                    && datafiles[del_idx]->seq_history == *(u32*)(&read_buf[buf_pos + 9])
                    && del_idx != old_idx)
                {
                    write(new_fd_key, &read_buf[buf_pos], cur_key_len + 13);
                    cur_key_pos += cur_key_len + 13;
                    new_df_ptr->count++;
                }
                else
                {
                    //skip and ignore
                }
                break;
            default:
                LOG_ERROR("Data file [%s] invalid.", tmp_path);
                eof = true;
                break;
            }
            buf_pos += cur_key_len + 13;
        }
        myfree(tmp_item);
        close(old_fd_key);
        close(old_fd_val);
        snprintf(tmp_path, PATH_MAX, "%s%09d.key", data_path, datafiles[old_idx]->seq);
        if (unlink(tmp_path))
        {
            LOG_WARN("Delete file %s error while compacting: %s", tmp_path, strerror(errno));
        }
        snprintf(tmp_path, PATH_MAX, "%s%09d.val", data_path, datafiles[old_idx]->seq);
        if (unlink(tmp_path))
        {
            LOG_WARN("Delete file %s error while compacting: %s", tmp_path, strerror(errno));
        }
        lock.WLock();
        myfree(datafiles[idx_list[i]]);
        datafiles[idx_list[i]] = NULL;
        if (i == 0) datafiles[new_idx] = new_df_ptr;
        lock.UnLock();
    }
    
    close(new_fd_key);
    close(new_fd_val);

    snprintf(tmp_path, PATH_MAX, "%s%09d.key", data_path, new_seq);
    snprintf(tmp_path2, PATH_MAX, "%s%09d.key.tmp", data_path, new_seq);
    if (rename(tmp_path2, tmp_path))
    {
        LOG_WARN("Rename file %s error while compacting: %s", tmp_path, strerror(errno));
    }
    else if (cur_key_pos == 8)
    {
        unlink(tmp_path);
    }

    snprintf(tmp_path, PATH_MAX, "%s%09d.val", data_path, new_seq);
    snprintf(tmp_path2, PATH_MAX, "%s%09d.val.tmp", data_path, new_seq);
    if (rename(tmp_path2, tmp_path))
    {
        LOG_WARN("Rename file %s error while compacting: %s", tmp_path, strerror(errno));
    }
    else if (cur_key_pos == 8)
    {
        unlink(tmp_path);
    }

    lock.WLock();
    if (cur_key_pos == 8) myfree(new_df_ptr);
    else datafiles[new_idx] = new_df_ptr;
    lock.UnLock();
    //TODO: Make the function more strong to tolerate exceptional interrupt.
}


void Bitcask::WriteOut()
{
    lock.WLock();
    write_out();
    lock.UnLock();
}

void Bitcask::fdisksync()
{
    if (cur_fd_key >= 0 && cur_fd_val >= 0)
    {
        fdatasync(cur_fd_val);
        fdatasync(cur_fd_key);
    }
}


void* Bitcask::aux_thread(void *instance)
{
    void *tret;
    int counter;
    Bitcask *bitcask = reinterpret_cast<Bitcask*>(instance);

    bitcask->compact_tid = 0;
    while(bitcask->is_running)
    {
        bitcask->WriteOut();
        bitcask->fdisksync();
        if (counter++ == 8)
        {
            if (bitcask->compact_status == 0 && bitcask->compact_tid == 0)
            {
                if (pthread_create(&(bitcask->compact_tid), NULL, Bitcask::compact, instance))
                {
                    LOG_ERROR("DB compaction thread create failed. msg: %s", strerror(errno));
                }

                //start_new_compact_thread
            }
            counter = 0;
        }
        if (bitcask->compact_status == 1)
        {
            pthread_join(bitcask->compact_tid, &tret);
            bitcask->compact_tid = 0;
            bitcask->compact_status = 0;
        }
        sleep(1);
    }

    if (bitcask->compact_status != 0)
    {
        pthread_join(bitcask->compact_tid, &tret);
        bitcask->compact_tid = 0;
    }
    return NULL;
}
